package net.bwie.realtime.warehouse.DWD;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.DWD
 * @Author: zhangleqing
 * @CreateTime: 2025-09-01  14:14
 * @Description: TODO dwd_Step_back（退中检测汇总表）
 * 相关表：refund_detail（退款明细信息表）  order_base（订单基础信息表）  product_info（商品信息表）
 * @Version: 1.0
 */
public class DwdStepBack {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 数据清洗+加工
        Table resultTable = handle(tableEnv);

//        resultTable.execute().print();

        // 4 映射表创建
        createView(tableEnv);

        // 5 输出数据
        saveToSink(tableEnv, resultTable);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("DwdStepBack", resultTable);
        tableEnv.executeSql(
                "insert into dwd_Step_back " +
                        "select * from DwdStepBack"
        ).print();
    }

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_Step_back (\n" +
                        "    product_category STRING,\n" +
                        "    product_id STRING,\n" +
                        "    \n" +
                        "    refund_id STRING,\n" +
                        "    refund_status STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    \n" +
                        "    order_id STRING,\n" +
                        "    order_amount DOUBLE,\n" +
                        "    pay_time TIMESTAMP(3)\n" +
                        ") WITH (\n" +
                        "     'connector' = 'kafka',\n" +
                        "     'topic' = 'dwd_Step_back',\n" +
                        "     'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "     'format' = 'json'\n" +
                        ")"
        );
    }


/*
select
    -- 商品信息
    p.product_category,
    p.product_id,
    -- 退款信息
    r.refund_id,
    r.refund_status,
    r.refund_amount,
    -- 订单信息
    o.order_id,
    o.order_amount,
    o.pay_time
from ods_refund_detail r
left join ods_order_base o
    ON r.order_id = o.order_id
left join ods_product_info as p
    on r.product_id = p.product_id

AND o.pay_time BETWEEN r.apply_time - INTERVAL '30' DAY AND r.apply_time + INTERVAL '30' DAY
含义：订单支付时间（pay_time）需在退款申请时间（apply_time）的前 30 天到后 30 天之间。

AND o.pay_time BETWEEN
        r.apply_time - INTERVAL '7' DAY - INTERVAL '1' HOUR  -- 提前1小时缓冲
        AND r.apply_time + INTERVAL '1' HOUR  -- 延后1小时缓冲
   -- 核心修正：支付时间必须在退款申请7天内（但申请时间肯定晚于支付时间）
   AND r.apply_time <= o.pay_time + INTERVAL '7' DAY


退中检测需监控超期退款：保留全量退款，新增字段
is_7day_rule（CASE WHEN r.apply_time <= o.pay_time + INTERVAL '7' DAY THEN 1 ELSE 0 END），
后续分析时按需过滤（如 WHERE is_7day_rule = 1 查合规退款，is_7day_rule = 0 查超期退款）；
 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "SELECT " +
                        "    p.product_category, " +
                        "    p.product_id, " +
                        "    r.refund_id, " +
                        "    r.refund_status, " +
                        "    r.refund_amount, " +
                        "    o.order_id, " +
                        "    o.order_amount, " +
                        "    o.pay_time " +
                        "FROM ods_refund_detail r " +  // 左表：退款明细表
                        "LEFT JOIN ods_order_base o "+  // 关联订单表 MINUTE
                        " ON r.order_id = o.order_id " +
//                        "   -- 核心修正：退款申请必须在订单支付后的 7 天内提交（含第 7 天当天）\n" +
//                        " AND r.apply_time <= o.pay_time + INTERVAL '7' DAY " +
                        // 关键修改：使用左表已有的事件时间属性apply_time关联维表
                        "LEFT JOIN ods_product_info FOR SYSTEM_TIME AS OF r.proctime p " +
                        "   ON r.product_id = p.product_id " +
                        "where o.order_id is not null and r.apply_time > o.pay_time" +
                        " and r.refund_id is not null and o.pay_time is not null"
        );
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        // 1 refund_detail（退款明细信息表）
        tableEnv.executeSql(
                "create table ods_refund_detail (\n" +
                        "    refund_detail_id STRING,\n" +
                        "    refund_id STRING,\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    apply_time TIMESTAMP(3),\n" +
                        "    refund_status STRING,\n" +
                        "    finish_time TIMESTAMP(3),\n" +
                        "    product_id STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    user_fill_reason STRING,\n" +
                        "    refund_scene STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR apply_time AS apply_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-refund-detail-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-refund-detail-log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 2 order_base（订单基础信息表）
        tableEnv.executeSql(
                "create table ods_order_base (\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    shop_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "    click_ship_time TIMESTAMP(3),\n" +
                        "    order_amount DECIMAL(10,2),\n" +
                        "    order_status STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-order-base-Log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-order-base-Log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 3 product_info（商品信息表）
        tableEnv.executeSql(
                "CREATE TABLE ods_product_info (\n" +
                        "    product_id STRING,\n" +
                        "    product_category STRING,\n" +
                        "    product_sku STRING,\n" +
                        "    product_price DOUBLE,\n" +
                        "    product_desc STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        // 新增：Doris FE 的 JDBC 端口（默认9030）
                        "    'jdbc-url' = 'jdbc:mysql://node102:9030',\n" +
                        "    'table.identifier' = 'transactions_ods.product_info',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        // 维表查询失败重试3次
                        "    'lookup.max-retries' = '3',\n"+
                        "    -- 移除不支持的'read.mode'参数（当前版本默认批处理模式）\n" +
                        "    'lookup.cache.max-rows' = '10000',  -- 支持的缓存参数\n" +
                        "    'lookup.cache.ttl' = '3600000'      -- 支持的缓存过期时间\n" +
                        ")"
        );
    }

    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
